package cn.com.registry;

import cn.com.common.PersistentInfo;
import cn.com.registry.etcd.EtcdClient;
import cn.com.registry.etcd.EtcdClientFactory;
import cn.com.common.RegistryInfo;
import com.alibaba.fastjson.JSON;
import com.ibm.etcd.api.Event;
import com.ibm.etcd.api.KeyValue;
import com.ibm.etcd.client.kv.KvClient;
import com.ibm.etcd.client.kv.WatchUpdate;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author jiaming
 */
public class EtcdRegistryService implements RegistryService {

    private static final Logger LOGGER = LogManager.getLogger();

    private EtcdClient etcdClient;
    private Notify notify;

    private static final ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(4, new ThreadFactory() {

        private AtomicInteger incrementAndGet = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("etcd-schedule-pool-" + incrementAndGet.incrementAndGet());
            return thread;
        }
    });

    public EtcdRegistryService(String url) {
        this.etcdClient = EtcdClientFactory.build(url);
    }

    @Override
    public void register(RegistryInfo registryInfo) {
        //check appName params
        if (registryInfo == null) {
            throw new IllegalArgumentException("registryInfo is not null");
        }
        String root = registryInfo.getRoot();
        String appName = registryInfo.getAppName();
        String host = registryInfo.getHost();
        int port = registryInfo.getPort();
        String registryKey = appendAppsRegistryKey(root, appName, host, port);
        doRegistry(registryKey, "");
    }

    @Override
    public void registerPersistent(RegistryInfo registryInfo) {
        if (registryInfo == null) {
            throw new IllegalArgumentException("registryInfo is not null");
        }
        if (registryInfo instanceof PersistentInfo) {
            PersistentInfo persistentInfo = (PersistentInfo) registryInfo;
            String jsonValues = "";
            if (persistentInfo.getValues() != null) {
                jsonValues = JSON.toJSONString(persistentInfo.getValues());
            }
            createPersistent(persistentInfo.getPath(), jsonValues);
        }
    }

    /**
     * 持久节点注册
     *
     * @param persistentKey
     * @param value
     */
    private void createPersistent(String persistentKey, String value) {
        etcdClient.put(persistentKey, value);
    }


    private void doRegistry(String registryKey, String value) {
        if (!etcdClient.checkPathExists(registryKey)) {
            //创建key,设置自动续租约
            try {
                etcdClient.keepAlive(registryKey, value, 5, 8);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
    }

    /**
     * @param appName
     * @param host
     * @param port
     * @return
     */
    private String appendAppsRegistryKey(String root, String appName, String host, int port) {
        StringBuilder sb = new StringBuilder();
        sb.append(root);
        sb.append("/");
        sb.append(appName);
        sb.append("/");
        sb.append(host);
        sb.append(":");
        sb.append(port);
        return sb.toString();
    }

    /**
     * 监听节点变更
     *
     * @param watchKeyPrefix
     */
    public void addListener(String watchKeyPrefix) {
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            KvClient.WatchIterator watchIterator = etcdClient.watchPrefix(watchKeyPrefix);
            while (watchIterator.hasNext()) {
                WatchUpdate next = watchIterator.next();
                List<Event> events = next.getEvents();
                if (!events.isEmpty()) {
                    List<RegistryInfo> removeRegInfos = new ArrayList<>();
                    List<RegistryInfo> putRegInfos = new ArrayList<>();
                    for (Event event : events) {
                        if (event.getType() == Event.EventType.DELETE) {
                            String key = event.getKv().getKey().toStringUtf8();
                            RegistryInfo registryInfo = toRegisterInfo(key);
                            removeRegInfos.add(registryInfo);
                        }
                        if (event.getType() == Event.EventType.PUT) {
                            String key = event.getKv().getKey().toStringUtf8();
                            RegistryInfo registryInfo = toRegisterInfo(key);
                            putRegInfos.add(registryInfo);
                        }
                    }
                    this.notify.notify(putRegInfos, removeRegInfos);
                }
            }
        }, 2, 5, TimeUnit.SECONDS);
    }

    protected RegistryInfo toRegisterInfo(String key) {
        if (key == null || key.length() == 0) {
            throw new IllegalArgumentException("toRegisterInfo fail,key is not null");
        }
        RegistryInfo registryInfo = new RegistryInfo();
        //key：/zhurong-apps/myapp/127.0.0.1:8080
        int i = key.indexOf("/", 1);
        if (i >= 0) {
            String root = key.substring(1, i);
            registryInfo.setRoot(root);
            key = key.substring(i);
        }
        i = key.indexOf("/", 1);
        if (i >= 0) {
            String appName = key.substring(1, i);
            registryInfo.setAppName(appName);
            key = key.substring(i + 1);
        }
        String host = key.substring(0, key.indexOf(":"));
        int port = Integer.parseInt(key.substring(key.indexOf(":") + 1));
        registryInfo.setHost(host);
        registryInfo.setPort(port);
        return registryInfo;
    }

    @Override
    public RegistryInfo subscribe(RegistryInfo registryInfo, String listenerPath) {
        if (registryInfo == null) {
            throw new IllegalArgumentException("registryInfo is not null");
        }
        if (!registryInfo.isListener()) {
            String registryKey = appendAppsRegistryKey(registryInfo.getRoot(),
                    registryInfo.getAppName(),
                    registryInfo.getHost(), registryInfo.getPort());
            //在etcd上注册client
            doRegistry(registryKey, "");
        }
        //拉取server端节点列表
        List<KeyValue> serverKeyValueList = etcdClient.getPrefix(listenerPath);
        if (serverKeyValueList != null && !serverKeyValueList.isEmpty()) {
            List<RegistryInfo> registryInfos = new ArrayList<>(serverKeyValueList.size());
            for (KeyValue keyValue : serverKeyValueList) {
                String key = keyValue.getKey().toStringUtf8();
                registryInfos.add(toRegisterInfo(key));
            }
            notify.notify(registryInfos, new ArrayList<>());
        }
        //监听server端节点变动
        this.addListener(listenerPath);
        return registryInfo;
    }

    @Override
    public void onListener(Notify notify) {
        setNotify(notify);
    }

    public void setNotify(Notify notify) {
        this.notify = notify;
    }
}
